package com.practice.car.cardataapp.schelduer.spark.task.sale;

import com.practice.car.cardataapp.schelduer.spark.Schedule;
import com.practice.car.cardataapp.schelduer.spark.analy.SaleRunner;
import net.sf.json.JSONArray;

import net.sf.json.JSONObject;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import scala.Tuple2;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.*;

public class MpvNationalSales {
    public static HashMap<String,Integer> getMPVNationalSales(){
        //1.将json文档转为JSONArray，注意这里的路径
        JSONArray dataJson = JSONArray.fromObject(readFile());
        List<String> list = new ArrayList<String>();

        //2.将JSONArray转为List
        for(int i=0; i<dataJson.size(); i++){
            JSONObject jo = dataJson.getJSONObject(i);
            String sale = jo.getString("sale");
            String national = jo.getString("national");
            String line ;
            String type = jo.getString("type");

            line = national + " " + sale;

            //筛选MPV系的数据
            if("MPV".equals(type.trim()))
                list.add(line);
       }
        //3.设置别名 和 运行模式
        SparkConf conf = new SparkConf().setAppName("MPVNationalSales").setMaster(Schedule.sparkMaster);
        //4.初始化
        JavaSparkContext sc = new JavaSparkContext(conf);
        //5.将list转为RDD
        JavaRDD dataRdd = sc.parallelize(list);
        //输出测试
//        dataRdd.foreach(new VoidFunction<String>() {
//            @Override
//          public void call(String o) throws Exception {
//               System.out.println(o);
//           }
//       });

        //6.将Rdd转为PairRdd，输入：String  输出:String  Integer
        JavaPairRDD dataPairRdd = dataRdd.mapToPair(new PairFunction<String,String,Integer>() {
            @Override
            public Tuple2 call(String line) throws Exception {
                String[] arr = line.split(" "); //拆分
                String national = new String(arr[0]);
                Integer sale;
                if("--".equals(arr[1])) //这里可能是数据有问题
                    sale=0;
                else
                    sale = Integer.parseInt(arr[1].trim());
                return new Tuple2(national,sale);
            }
        });

        //7.使用PairRdd进行reduce 输入，将相同national的sale相加
        JavaPairRDD resultPairRdd = dataPairRdd.reduceByKey(new Function2<Integer,Integer,Integer>() {
            @Override
            public Integer call(Integer sale1,Integer sale2) throws Exception {
                return sale1 + sale2;
            }
        });

//        //8.foreach 输出测试
//        resultPairRdd.foreach(new VoidFunction< Tuple2<String,Integer> >() {
//            @Override
//            public void call(Tuple2<String , Integer> t) throws Exception {
//                System.out.println(t._1 + ":" + t._2);
//            }
//        });

        //9.PairRdd转为Map
        Map<String,Integer> map =  resultPairRdd.collectAsMap();
        HashMap<String,Integer> hashMap = new HashMap<String,Integer>(map);

        //10.遍历HashMap遍历输出
        Set<Map.Entry<String,Integer>> entry=hashMap.entrySet();
        sc.close();
        return hashMap;
    }

    /**
     * 读取hdfs指定路径的内容
     */
    public static String readFile(){
        String hdfsPath = Schedule.saleHdfsUrl;
        String result = "";
        Path path = new Path(hdfsPath);
        Configuration configuration = new Configuration();
        FSDataInputStream fsDataInputStream = null;
        FileSystem fileSystem = null;
        BufferedReader br = null;
        // 定义一个字符串用来存储文件内容
        try {
            fileSystem = path.getFileSystem(configuration);
            fsDataInputStream = fileSystem.open(path);
            br = new BufferedReader(new InputStreamReader(fsDataInputStream));
            String str2;
            while ((str2 = br.readLine()) != null) {
                // 遍历抓取到的每一行并将其存储到result里面
                //str2 = decodeUnicode(str2);
                result += str2 + "\n";
            }
        } catch (IOException e) {
            e.printStackTrace();
        } finally {
            if(br!=null){
                try {
                    br.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if(fsDataInputStream!=null){
                try {
                    fsDataInputStream.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            if(fileSystem!=null){
                try {
                    fileSystem.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }

        return result;
    }


    //json数据解码
    public static String decodeUnicode(String theString) {
        char aChar;
        int len = theString.length();
        StringBuffer outBuffer = new StringBuffer(len);
        for (int x = 0; x < len;) {
            aChar = theString.charAt(x++);
            if (aChar == '\\') {
                aChar = theString.charAt(x++);
                if (aChar == 'u') {
                    // Read the xxxx
                    int value = 0;
                    for (int i = 0; i < 4; i++) {
                        aChar = theString.charAt(x++);
                        switch (aChar) {
                            case '0':
                            case '1':
                            case '2':
                            case '3':
                            case '4':
                            case '5':
                            case '6':
                            case '7':
                            case '8':
                            case '9':
                                value = (value << 4) + aChar - '0';
                                break;
                            case 'a':
                            case 'b':
                            case 'c':
                            case 'd':
                            case 'e':
                            case 'f':
                                value = (value << 4) + 10 + aChar - 'a';
                                break;
                            case 'A':
                            case 'B':
                            case 'C':
                            case 'D':
                            case 'E':
                            case 'F':
                                value = (value << 4) + 10 + aChar - 'A';
                                break;
                            default:
                                throw new IllegalArgumentException(
                                        "Malformed   \\uxxxx   encoding.");
                        }

                    }
                    outBuffer.append((char) value);
                } else {
                    if (aChar == 't')
                        aChar = '\t';
                    else if (aChar == 'r')
                        aChar = '\r';
                    else if (aChar == 'n')
                        aChar = '\n';
                    else if (aChar == 'f')
                        aChar = '\f';
                    outBuffer.append(aChar);
                }
            } else
                outBuffer.append(aChar);
        }
        return outBuffer.toString();
    }

}
